/**
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.ubox.mqtt_upush.config;

import java.io.File;
import java.net.URI;
import java.util.ArrayList;
import java.util.LinkedList;
import java.util.List;

import javax.annotation.PreDestroy;

import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.broker.BrokerPlugin;
import org.apache.activemq.broker.BrokerService;
import org.apache.activemq.broker.TransportConnector;
import org.apache.activemq.store.kahadb.KahaDBStore;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.io.ClassPathResource;

import com.ubox.mqtt_upush.mqttbroker.ResourceLoadingSslContext;
import com.ubox.mqtt_upush.mqttbroker.SessionFilter;
import com.ubox.mqtt_upush.mqttbroker.SessionPlugin;
@Configuration
@ConfigurationProperties(prefix = "mqtt")
public class MqttBrokerConfig {

	private static final Logger LOG = LoggerFactory.getLogger(MqttBrokerConfig.class);

	public static final String KAHADB_DIRECTORY = "activemq-data/";

	protected BrokerService brokerService;
	SessionFilter sessionFilter;
	protected int port;
	protected int sslPort;
	protected String host = "localhost";
	protected String jmsUri = "vm://localhost:16161";
	protected ActiveMQConnectionFactory cf;
	protected LinkedList<Throwable> exceptions = new LinkedList<Throwable>();
	protected boolean persistent;
	protected String protocolConfig;
	protected String protocolScheme;
	protected boolean useSSL = true;
	private String keyStore;
	private String keyStorePassword;

	public static final int AT_MOST_ONCE = 0;
	public static final int AT_LEAST_ONCE = 1;
	public static final int EXACTLY_ONCE = 2;

	public MqttBrokerConfig() {
		this.protocolScheme = "mqtt+nio";
		this.useSSL = true;
	}

	public MqttBrokerConfig(String connectorScheme, boolean useSSL) {
		this.protocolScheme = connectorScheme;
		this.useSSL = useSSL;
	}

	public static String getTestName() {
		return "ubox";
	}

	@Bean
	public SessionFilter brokerService() throws Exception {

		this.protocolScheme = "mqtt+nio+ssl";
		
		exceptions.clear();
		
		brokerService = createBroker(true);

		sessionFilter = configurePlugins(brokerService);

		brokerService.start();
		brokerService.waitUntilStarted();
		port = brokerService.getTransportConnectorByName("mqtt+nio").getConnectUri().getPort();
		jmsUri = brokerService.getTransportConnectorByName("openwire").getPublishableConnectString();
		cf = new ActiveMQConnectionFactory(jmsUri);
		
		return sessionFilter;
	}

	protected BrokerService createBroker(boolean deleteAllMessages) throws Exception {
		BrokerService broker = new BrokerService();
		broker.setDeleteAllMessagesOnStartup(deleteAllMessages);
		broker.setPersistent(isPersistent());
		if (isPersistent()) {
			KahaDBStore kaha = new KahaDBStore();
			kaha.setDirectory(new File(KAHADB_DIRECTORY + getTestName()));
			broker.setPersistenceAdapter(kaha);
		}
		broker.setAdvisorySupport(false);
		broker.setUseJmx(true);
		broker.getManagementContext().setCreateConnector(false);
		broker.setSchedulerSupport(isSchedulerSupportEnabled());
		broker.setPopulateJMSXUserID(true);

		applyBrokerPolicies(broker);
		applyMemoryLimitPolicy(broker);

		addMQTTConnector(broker);//mqtt tcp 1883
		
		addWSConnector(broker);//

		// Setup SSL context...
		org.springframework.core.io.Resource fileRource = new ClassPathResource(keyStore);
		final ResourceLoadingSslContext sslContext = new ResourceLoadingSslContext();
		sslContext.setKeyStore(fileRource.getURI().toString());
		LOG.info("keyStore uri: " + fileRource.getURI().toString());
		sslContext.setKeyStorePassword(keyStorePassword);
		sslContext.afterPropertiesSet();
		broker.setSslContext(sslContext);

		addMQTTSSLConnector(broker);// 8883
		
		addOpenWireConnector(broker);//18883

		return broker;
	}

	private void addWSConnector(BrokerService broker) throws Exception {

		String url = "ws://0.0.0.0:" + 18883 + "?allowLinkStealing=" + isAllowLinkStealing()
				+ "&websocket.maxTextMessageSize=99999" + "&transport.idleTimeout=1001"
				+ "&trace=true&transport.trace=true";
		broker.addConnector(url);
	}
	
	protected boolean isAllowLinkStealing() {
		return true;
	}

	protected SessionFilter configurePlugins(BrokerService brokerService) throws Exception {
		ArrayList<BrokerPlugin> plugins = new ArrayList<BrokerPlugin>();
		createPlugins(plugins);
		SessionPlugin sessionPlugin = new SessionPlugin();
		plugins.add(sessionPlugin);

		BrokerPlugin authenticationPlugin = configureAuthentication();
		if (authenticationPlugin != null) {
			plugins.add(configureAuthorization());
		}

		BrokerPlugin authorizationPlugin = configureAuthorization();
		if (authorizationPlugin != null) {
			plugins.add(configureAuthentication());
		}

		if (!plugins.isEmpty()) {
			BrokerPlugin[] array = new BrokerPlugin[plugins.size()];
			brokerService.setPlugins(plugins.toArray(array));
		}
		return sessionPlugin.sessionFilter;
	}

	/**
	 * Allows a subclass to add additional broker plugins during the broker startup
	 * process. This method should not add Authorization or Authentication plugins
	 * as those are handled by the configureAuthentication and
	 * configureAuthorization methods later.
	 *
	 *
	 *
	 * @param plugins
	 *            The List object to add Plugins for installation into the new
	 *            Broker.
	 *
	 * @throws Exception
	 *             if an error occurs during the plugin creation process.
	 */
	protected void createPlugins(List<BrokerPlugin> plugins) throws Exception {

	}

	protected BrokerPlugin configureAuthentication() throws Exception {
		return null;
	}

	protected BrokerPlugin configureAuthorization() throws Exception {
		return null;
	}

	protected void applyBrokerPolicies(BrokerService brokerService) throws Exception {
		// NOOP here
	}

	protected void applyMemoryLimitPolicy(BrokerService brokerService) throws Exception {
	}

	protected void addOpenWireConnector(BrokerService brokerService) throws Exception {
		TransportConnector connector = new TransportConnector();
		connector.setUri(new URI("tcp://0.0.0.0:0"));
		connector.setName("openwire");
		brokerService.addConnector(connector);
	}

	protected void addMQTTConnector(BrokerService brokerService) throws Exception {
		// Overrides of this method can add additional configuration options or add
		// multiple
		// MQTT transport connectors as needed, the port variable is always supposed to
		// be
		// assigned the primary MQTT connector's port.
		String url = "mqtt+nio";
		StringBuilder connectorURI = new StringBuilder();
		connectorURI.append(url);
		connectorURI.append("://0.0.0.0:").append(port);
		String protocolConfig = getProtocolConfig();
		if (protocolConfig != null && !protocolConfig.isEmpty()) {
			connectorURI.append("?").append(protocolConfig);
		}

		TransportConnector connector = new TransportConnector();
		connector.setUri(new URI(connectorURI.toString()));
		connector.setName(url);
		brokerService.addConnector(connector);

		LOG.info("Added connector {}:{} to broker", url, port);
	}

	private void addMQTTSSLConnector(BrokerService brokerService) throws Exception {
		// Overrides of this method can add additional configuration options or add
		// multiple
		// MQTT transport connectors as needed, the port variable is always supposed to
		// be
		// assigned the primary MQTT connector's port.

		StringBuilder connectorURI = new StringBuilder();
		connectorURI.append(getProtocolScheme());
		connectorURI.append("://0.0.0.0:").append(sslPort);
		String protocolConfig = getProtocolConfig();
		if (protocolConfig != null && !protocolConfig.isEmpty()) {
			connectorURI.append("?").append(protocolConfig);
		}
		//
		TransportConnector connector = new TransportConnector();
		connector.setUri(new URI(connectorURI.toString()));
		connector.setName("mqtt+nio+ssl");
		brokerService.addConnector(connector);
		LOG.info("Added connector {}:{} to broker", getProtocolScheme(), sslPort);

	}

	public void stopBroker() throws Exception {
		if (brokerService != null) {
			brokerService.stop();
			brokerService.waitUntilStopped();
			brokerService = null;
		}
	}
	
	@PreDestroy
	public void destory() throws Exception {  
		stopBroker();
		LOG.info("成功关闭broker");
    }  

	public String getProtocolScheme() {
		return protocolScheme;
	}

	public void setProtocolScheme(String scheme) {
		this.protocolScheme = scheme;
	}

	public String getProtocolConfig() {
		return protocolConfig;
	}

	public void setProtocolConfig(String config) {
		this.protocolConfig = config;
	}

	public boolean isPersistent() {
		return persistent;
	}

	public int getPort() {
		return this.port;
	}

	public boolean isSchedulerSupportEnabled() {
		return false;
	}

	public String getHost() {
		return host;
	}

	public void setHost(String host) {
		this.host = host;
	}

	public void setPort(int port) {
		this.port = port;
	}

	public void setPersistent(boolean persistent) {
		this.persistent = persistent;
	}

	public int getSslPort() {
		return sslPort;
	}

	public void setSslPort(int sslPort) {
		this.sslPort = sslPort;
	}

	public String getKeyStorePassword() {
		return keyStorePassword;
	}

	public void setKeyStorePassword(String keyStorePassword) {
		this.keyStorePassword = keyStorePassword;
	}

	public String getKeyStore() {
		return keyStore;
	}

	public void setKeyStore(String keyStore) {
		this.keyStore = keyStore;
	}

}
